package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.common.ConstantTopic;
import com.atguigu.realtime.util.*;
import io.debezium.data.Json;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: DWMOrderWideApp
 * @Description: V2.0     将数据缓存在redis中
 * @Author: kele
 * @Date: 2021/4/20 14:53
 *
 * 升级说明：
 *         针对v2版本，数据请求redis需要通过网络，每条数据必须等待连接，处理，返回数据之后才能进行下一条数据处理
 *
 *         在异步模式下，将IO操作异步化，单个并行可以连续发送多个请求，哪个请求先返回就先处理，
 *         从而在连续的请求间不需要阻塞式等待，大大提高了流处理效率。
 *
 *  flink在1.2时支持异步API：AsyncDataStream
 *
 *  AsyncDataStream使用说明：
 *  AsyncDataStream.unorderedWait(orderWideStream,
 *                 new RichAsyncFunction<OrderWide, OrderWide>() {
 *
 * 					private Jedis redis;
 *
 *                     @Override
 *                     public void open(Configuration parameters) throws Exception {
 *
 *                         //获取线程池
 *                     }
 *
 *                     //处理函数
 *                     @Override
 *                     public void asyncInvoke(OrderWide input, ResultFuture<OrderWide> resultFuture) throws Exception {
 *
 *                         pool.execute(new Runnable() {
 *
 *                             @Override
 *                             public void run() {
 *                                 //获取jedis连接
 *                                 //执行的代码
 *                                 //关闭jedis连接
 *                             }
 *                         });
 *                     }
 *
 *                     @Override
 *                     public void close() throws Exception {
 *                         //关闭线程池
 *                 },
 *                 30,
 *                 TimeUnit.SECONDS
 *         )}
 **/
public class DWMOrderWideAppV3 extends BaseAppV2 {

    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        new DWMOrderWideAppV3().init(30010,
                2,
                "DWMOrderWideApp",
                "DWMOrderWideApp",
                ConstantTopic.DWD_ORDER_INFO,
                ConstantTopic.DWD_ORDER_DETAIL);
    }

    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStream<String>> streams) throws SQLException, ClassNotFoundException {

        //1、将订单表和订单详情表进行interval join（两个事实表）
        SingleOutputStreamOperator<OrderWide> orderWideStream = StreamJoin(streams);

        //2、与维度表进行join，通过查询phoenix进行join
        SingleOutputStreamOperator<OrderWide> joinDimStream = JoinDim(orderWideStream);

    //    joinDimStream.print();

        //3、将数据发送到kafka
        send2Kafka(joinDimStream);

    }

    private void send2Kafka(SingleOutputStreamOperator<OrderWide> joinDimStream) {

        joinDimStream
                .map(JSONObject::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(ConstantTopic.DWM_ORDER_WIDE));

    }

    /**
     * 与维度表进行关联，增加redis缓存
     *
     * @param orderWideStream
     */
    private SingleOutputStreamOperator<OrderWide> JoinDim(SingleOutputStreamOperator<OrderWide> orderWideStream) throws SQLException, ClassNotFoundException {

        //通过异步API处理，
        SingleOutputStreamOperator<OrderWide> result = AsyncDataStream.unorderedWait(orderWideStream,
                new RichAsyncFunction<OrderWide, OrderWide>() {

                    private ThreadPoolExecutor pool;
                    private Connection connect;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connect = MyJDBCUtil.getConnect(Constant.PHOENIX_URL, Constant.PHOENIX_DRIVER);
                        pool = MyThreadPooluUtil.getPool();
                    }

                    //处理函数
                    @Override
                    public void asyncInvoke(OrderWide orderWide, ResultFuture<OrderWide> resultFuture) throws Exception {

                        pool.execute(new Runnable() {

                            //将数据执行代码放入.......
                            @SneakyThrows
                            @Override
                            public void run() {
                                Jedis redis = MyJedisUtil.getRedis();

                                JSONObject userInfo = DimQuery
                                        .getDimQuery(connect, redis, "DIM_USER_INFO", orderWide.getUser_id());
                                //在phoenix中，查询出的字段名是大写的
                                orderWide.setUser_gender(userInfo.getString("GENDER"));
                                orderWide.calcAge(userInfo.getString("BIRTHDAY"));

                                //关联dim_base_province表
                                JSONObject baseProvince = DimQuery
                                        .getDimQuery(connect, redis, "DIM_BASE_PROVINCE", orderWide.getProvince_id());
                                orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
                                orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                                orderWide.setProvince_name(baseProvince.getString("NAME"));
                                orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));

                                //关联sku_info表
                                JSONObject skuInfo = DimQuery
                                        .getDimQuery(connect, redis, "DIM_SKU_INFO", orderWide.getSku_id());
                                orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                                orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                                orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                                orderWide.setTm_id(skuInfo.getLong("TM_ID"));

                                //关联spu_info表
                                JSONObject spuInfo = DimQuery
                                        .getDimQuery(connect, redis, "DIM_SPU_INFO", orderWide.getSpu_id());
                                orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                                //关联dim_base_category3表
                                JSONObject category3 = DimQuery
                                        .getDimQuery(connect, redis, "DIM_BASE_CATEGORY3", orderWide.getCategory3_id());
                                orderWide.setCategory3_name(category3.getString("NAME"));

                                //关联DIM_BASE_TRADEMARK表
                                JSONObject baseTrademark = DimQuery
                                        .getDimQuery(connect, redis, "DIM_BASE_TRADEMARK", orderWide.getTm_id());
                                orderWide.setTm_name(baseTrademark.getString("TM_NAME"));

                                redis.close();

                                //orderwide是一条数据，complete接收一个集合，将orderwide封装为单个list集合
                                //将数据返回
                                resultFuture.complete(Collections.singletonList(orderWide));
                            }
                        });
                    }

                    @Override
                    public void close() throws Exception {
                        if (connect != null) {
                            connect.close();
                        }

                        if (pool != null) {
                            pool.shutdown();
                        }
                    }
                },
                30,
                TimeUnit.SECONDS
        );

        return result;


    }

    /**
     * 将订单表和订单详情表进行join，得到宽表
     * <p>
     * 使用interval join的方式进行join，
     * --1、需要使用事件时间
     * --2、必须是keyby之后的数据
     */
    private SingleOutputStreamOperator<OrderWide> StreamJoin(Map<String, DataStream<String>> streams) {

        //获取数据，封装为对象，添加水印，按照订单id进行keyby

        KeyedStream<OrderInfo, Long> orderInfoStream = streams
                .get("dwd_order_info")
                .map(s -> JSON.parseObject(s, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);


        KeyedStream<OrderDetail, Long> orderDetailStream = streams
                .get("dwd_order_detail")
                .map(s -> JSON.parseObject(s, OrderDetail.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((ele, ts) -> ele.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);

        SingleOutputStreamOperator<OrderWide> result = orderInfoStream
                .intervalJoin(orderDetailStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo,
                                               OrderDetail orderDetail,
                                               Context ctx,
                                               Collector<OrderWide> out) throws Exception {

                        //将两个流join的结果作为OrderWide输出
                        out.collect(new OrderWide(orderInfo, orderDetail));
                    }
                });

        return result;
    }
}

